package com.kafka_test.app.service;

import com.alibaba.fastjson.JSONObject;
import com.kafka_test.app.mapper.OffsetMapper;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Service;

import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.regex.Pattern;

@Service
public class OffsetService {
    private final KafkaTemplate<Integer, String> kafkaTemplate;
    private static OffsetMapper offsetMapper;

    @Autowired
    public OffsetService(OffsetMapper offsetMapper, KafkaTemplate<Integer, String> kafkaTemplate) {
        OffsetService.offsetMapper = offsetMapper;
        this.kafkaTemplate=kafkaTemplate;
    }

    public List<Map<String, Object>> queryTableData(String tableName, long offset) {
        return offsetMapper.queryTableData(tableName, offset);
    }

    public Object getOffset(String tableName) {
        return offsetMapper.getOffset(tableName);
    }

    public void initializeOffset(String tableName, long offset) {
        offsetMapper.initializeOffset(tableName, offset);
    }

    public void sendMessage(List<String> resultList, String topic) {
        for(String data : resultList)
            kafkaTemplate.send(topic,0, data);
    }

    public void updateOffset(String tableName, long offset) {
        offsetMapper.updateOffset(tableName, offset);
    }

    public void insertConsumer(String tableName, String data){
        JSONObject json = JSONObject.parseObject(data);
        Set<String> keys = json.keySet();
        for(String key : keys){
            if(Pattern.matches("id_.*", key))
                continue;
            offsetMapper.insertConsumer(tableName, key, json.get(key));
        }
    }
}
